package com.xiaoxu.yihaoShop.ETLJob;

import com.xiaoxu.utils.GetPageId;
import com.xiaoxu.utils.LogParser;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Map;

public class ETLMap extends Mapper<LongWritable, Text, NullWritable, Text> {

    private LogParser logParser;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        logParser = new LogParser();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        Map<String, String> logInfo = logParser.parse(value.toString());

        String ip = logInfo.get("ip");
        String url = logInfo.get("url");
        String sessionId = logInfo.get("sessionId");
        String time = logInfo.get("time");
        String country = logInfo.get("country") == null ? "-" : logInfo.get("country");
        String province = logInfo.get("province") == null ? "-" : logInfo.get("province");
        String city = logInfo.get("city") == null ? "-" : logInfo.get("city");
        String pageId = GetPageId.getPageId(url) == "" ? "-" : GetPageId.getPageId(url);

        StringBuilder builder = new StringBuilder();
        builder.append(ip).append("\t");
        builder.append(country).append("\t");
        builder.append(province).append("\t");
        builder.append(city).append("\t");
        builder.append(sessionId).append("\t");
        builder.append(time).append("\t");
        builder.append(pageId).append("\t");
        builder.append(url);


        if (StringUtils.isNotBlank(pageId) && !pageId.equals("-")) {
            System.out.println("------" + pageId);
        }


        context.write(NullWritable.get(), new Text(builder.toString()));

    }
}
